/*!
 shared handle for axum web handler

*/

use crate::*;
use rabbitmq::client::ChannelClient;
use tokio::sync::{
    mpsc::{Receiver, Sender},
    Mutex, RwLock,
};
// use std::collections::HashMap;
/**
 * 维护共享资源的管理器
 *
 */

pub struct ShareComm {
    pub tx: Sender<Task>,
    pub rx: Mutex<Receiver<Task>>,
    pub ptx: Sender<ParaTask>,
    pub prx: Mutex<Receiver<ParaTask>>,
    // gfs没有做锁同步，sqlite具备解决能力。
    pub gfs: Gfs,
    pub index: NodeIndex,
    pub token: String,
    /// 用于同步消息的MQ服务器， rabbitmq:  ip:port;  redis:  redis://passwd@ip:port/
    pub mq_server: String,
    pub sync_id: Mutex<i32>,
    pub record: RwLock<TaskStatus>,
    pub port: u16,
    ///////////////
    pub rmq_client: Arc<ChannelClient>,
}

impl ShareComm {
    pub fn new(
        gfs: Gfs,
        index: NodeIndex,
        mq: &str,
        token: &str,
        web_port: u16,
        rmqc: Arc<ChannelClient>,
    ) -> Self {
        // 256 个节点，可以一次性缓存每个节点的信息。
        let (tx, rx) = tokio::sync::mpsc::channel(256);
        let (ptx, prx) = tokio::sync::mpsc::channel(2);
        let record = TaskStatus::new();
        Self {
            rx: tokio::sync::Mutex::new(rx),
            tx,
            prx: tokio::sync::Mutex::new(prx),
            ptx,
            gfs,
            index,
            mq_server: mq.to_string(),
            token: token.to_string(),
            sync_id: Mutex::new(0),
            port: web_port,
            record: RwLock::new(record),
            rmq_client: rmqc,
        }
    }

    pub async fn get_next_id(&self) -> i32 {
        let id = self.gfs.get_last_id().await;
        id + 1
    }

    pub async fn get_next_from_id(&self, id: i32, limit: u32) -> Result<Vec<Stat>> {
        let re = self.gfs.get_next_from_id(id, limit).await;
        re
    }

    pub async fn get_by_name(
        &self,
        hash: &str,
        fname: &str,
        id: i32,
        limit: u32,
    ) -> Result<Vec<Stat>> {
        let re = self.gfs.get_by_name(hash, fname, id, limit).await;
        re
    }

    pub fn is_master(&self) -> bool {
        return self.gfs.master.len() == 0;
    }
    pub async fn get_slaves(&self) -> Option<Vec<String>> {
        if self.mq_server.len() > 0 && !self.mq_server.starts_with("redis") && self.is_master() {
            return self.rmq_client.subclients().await;
        }
        // 这里，可以考虑提交给work，work通过 MqActor去执行远端cache
        error!("logic error, this is a slave, cant use rabbitmq");
        return None;
    }
}
//////////
